package com.deep.flink.demo.job._2_join;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class _1_MySqlJoinJob {

    public static void main(String[] args) {
        //
        StreamExecutionEnvironment exeEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        //========== 【表环境】 ==========
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(exeEnv, settings);
        //========== 【定义表】 ==========
        createOrderBaseDDL(tableEnv);
        createUserBaseDDL(tableEnv);
        createSinkDDL(tableEnv);
        //========== 【Join】 ==========
        Table table = tableEnv.sqlQuery(sql());
        StatementSet st = tableEnv.createStatementSet();
        st.addInsert("order_base_detail", table);
        //========== 【执行】 ==========
        st.execute();
    }

    public static void createOrderBaseDDL(StreamTableEnvironment tableEnv) {
        //
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE order_base (")
                .append("rb_id      BIGINT, ")
                .append("rb_ub_id   BIGINT, ")
                .append("rb_amount  STRING, ")
                .append("PRIMARY KEY (rb_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'mysql-cdc', ")
                .append("'hostname' = '127.0.0.1', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'port' = '3306', ")
                .append("'database-name' = 'ds_canal', ")
                .append("'table-name'= 'order_base', ")
                .append("'server-id' = '333', ")
                .append("'server-time-zone' = 'Asia/Shanghai', ")
                .append("'debezium.database.whitelist' = 'ds_canal', ")
                .append("'debezium.table.whitelist' = 'ds_canal.order_base', ")
                .append("'debezium.include.schema.changes' = 'false' ")
                .append(")");
        //
        TableResult tblResult = tableEnv.executeSql(ddl.toString());
    }

    public static void createUserBaseDDL(StreamTableEnvironment tableEnv) {
        //
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE user_base (")
                .append("ub_id          BIGINT, ")
                .append("ub_login_name  STRING, ")
                .append("ub_login_pwd   STRING, ")
                .append("ub_begin_time  TIMESTAMP, ")
                .append("PRIMARY KEY (ub_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'mysql-cdc', ")
                .append("'hostname' = '127.0.0.1', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'port' = '3306', ")
                .append("'database-name' = 'ds_canal', ")
                .append("'table-name'= 'user_base', ")
                .append("'server-id' = '444', ")
                .append("'server-time-zone' = 'Asia/Shanghai', ")
                .append("'debezium.database.whitelist' = 'ds_canal', ")
                .append("'debezium.table.whitelist' = 'ds_canal.user_base', ")
                .append("'debezium.include.schema.changes' = 'false' ")
                .append(")");
        //
        TableResult tblResult = tableEnv.executeSql(ddl.toString());
    }

    public static void createSinkDDL(StreamTableEnvironment tableEnv) {
        //
        StringBuffer ddl = new StringBuffer();
        ddl.append("CREATE TABLE order_base_detail (")
                .append("rb_id          BIGINT, ")
                .append("rb_amount      STRING, ")
                .append("ub_id          BIGINT, ")
                .append("ub_login_name  STRING, ")
                .append("PRIMARY KEY (rb_id) NOT ENFORCED")
                .append(") WITH (")
                .append("'connector' = 'jdbc', ")
                .append("'url' = 'jdbc:mysql://127.0.0.1:3306/ds_canal?serverTimezone=UTC', ")
                .append("'username' = 'root', ")
                .append("'password' = 'root', ")
                .append("'table-name' = 'order_base_detail', ")
                .append("'driver' = 'com.mysql.jdbc.Driver' ")
                .append(")");
        //
        TableResult tblResult = tableEnv.executeSql(ddl.toString());
    }

    public static String sql() {
        StringBuffer sql = new StringBuffer();
        sql.append("select rb.rb_id          as rb_id, ")
                .append("  rb.rb_amount      as rb_amount, ")
                .append("  rb.rb_ub_id       as ub_id, ")
                .append("  ub.ub_login_name  as ub_login_name ")
                .append("from order_base rb ")
                .append("left join user_base ub on rb.rb_ub_id = ub.ub_id ");
        return sql.toString();
    }
}


